package cn.doitedu.demo1;


import cn.doitedu.beans.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/12
 * @Desc: 学大数据，上多易教育
 * <p>
 * 静态画像条件（画像库中有现成结果的）：
 * 职业：宝妈，
 * 年龄：20-30
 * 近半年平均月消费额： 1000-1200
 * <p>
 * 动态画像统计条件：
 * A行为 3+ 次
 * [B行为，E行为 ，Q行为] 序列 2+ 次
 * <p>
 * 规则触发条件：
 * 当她浏览母婴用品时触发推送消息
 **/
public class EntryPoint {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造kafkaSource
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("ods-user-action-log")
                .setGroupId("ggg")
                .setClientIdPrefix("client-001")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(json -> JSON.parseObject(json, EventBean.class));

        SingleOutputStreamOperator<String> resultStream = beanStream.keyBy(EventBean::getUser_id)
                .process(new KeyedProcessFunction<Long, EventBean, String>() {

                    RestHighLevelClient client;
                    SearchRequest request;
                    BoolQueryBuilder boolQueryBuilder;

                    ValueState<Integer> cntState;

                    ValueState<Integer> idxState;
                    ValueState<Integer> seqCntState;

                    String[] actions = {"B", "E", "Q"};

                    JSONObject message;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        // 先构造客户端对象
                        client = new RestHighLevelClient(RestClient.builder(new HttpHost("doitedu", 9200, "http")));


                        // 构造请求对象
                        request = new SearchRequest("profile_test");


                        // 构造职业标签查询条件
                        MatchQueryBuilder job = QueryBuilders.matchQuery("tag_01_01", "宝妈");
                        // 构造年龄标签查询条件
                        RangeQueryBuilder age = QueryBuilders.rangeQuery("tag_01_02").gte(20).lte(30);
                        // 构造月平均消费额标签查询条件
                        RangeQueryBuilder cost = QueryBuilders.rangeQuery("tag_03_01").gte(1000).lte(1200);


                        boolQueryBuilder = QueryBuilders.boolQuery();
                        boolQueryBuilder.must(job).must(age).must(cost);

                        cntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("rule-1-1-cnt", Integer.class));

                        idxState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("rule-1-1-idx", Integer.class));
                        seqCntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("rule-1-1-seq", Integer.class));

                        message = new JSONObject();
                        message.put("rule_id","rule-1-1");


                    }

                    @Override
                    public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context ctx, Collector<String> out) throws Exception {


                        long userId = eventBean.getUser_id();
                        String eventId = eventBean.getEvent_id();

                        /**
                         * 一、 查询es，判断当前用户是否满足本规则的静态画像条件
                         */

                        IdsQueryBuilder id = QueryBuilders.idsQuery().addIds(String.valueOf(userId));

                        boolQueryBuilder.must(id);
                        request.source(new SearchSourceBuilder().query(boolQueryBuilder));

                        // 用客户端发起请求
                        SearchResponse response = client.search(request, RequestOptions.DEFAULT);

                        // 解析结果
                        SearchHits hits = response.getHits();
                        // 取出返回的结果条数
                        long value = hits.getTotalHits().value;
                        // 如果根据上述的条件组合查不到任何结果，则说明当前用户不满足规则的静态画像条件
                        if (value < 1) return;

                        System.out.println("es查询通过.................");


                        /**
                         * 二、 对规则的动态画像统计条件进行处理
                         */
                        if ("A".equals(eventId)) {
                            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);

                            System.out.println("更新了A的次数： ................." + cntState.value());
                        }

                        // String[] actions = {"B", "E", "Q"};
                        int idx = idxState.value() == null ? 0 : idxState.value();
                        if(eventId.equals(actions[idx])){
                            idx++;
                            if(idx == actions.length ){
                                idx = 0;
                                seqCntState.update(seqCntState.value() == null ? 1 : seqCntState.value() + 1);

                                System.out.println("更新了序列的次数： ................." + seqCntState.value());


                            }
                            idxState.update(idx);

                        }

                        /**
                         * 三、 判断当前行为是否是本规则的触发条件事件
                         */
                        if("X".equals(eventId)){

                            System.out.println("触发行为到达：此时的状态数据: " + cntState.value() + "," + seqCntState.value());

                            if(cntState.value() != null && cntState.value() >= 3 && seqCntState.value() != null && seqCntState.value() >= 2){

                                message.put("user_id",userId);
                                message.put("hit_time",eventBean.getAction_time());

                                out.collect(message.toJSONString());
                            }

                        }


                        /**
                         * 开始对第二个规则进行处理
                         */








                        /**
                         * 开始对第三个规则进行处理
                         */





                    }
                });


        resultStream.print();

        env.execute("我爱你");


    }
}
